package com.powernode.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.powernode.EsDomin.ProdEs;
import com.powernode.constant.QueueConstant;
import com.powernode.dto.EsChange;
import com.powernode.service.PropRepository;
import com.powernode.domain.Prod;
import com.powernode.service.EsImportService;
import com.powernode.service.ProdService;
import com.powernode.util.ProdThreadUtil;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;

@Service
public class EsImportServiceImpl implements EsImportService, CommandLineRunner {

    //获取每次导入的数量
    @Value("${esimport.size}")
    private Integer size;

    //注入接口
    @Autowired
    private PropRepository propRepository;

    @Autowired
    private ProdService prodService;

    private Date start;

    /**
     * 上传全部到es中
     * 导入es的包
     * 配置文件配置路径
     * 创建实体类
     * 创建接口并继承
     */
    @Override
    public void importAll() {
        //获取商品总数
        int total = prodService.count(new LambdaQueryWrapper<Prod>()
                .eq(Prod::getStatus, 0));
        //得到操作次数
        int num = total / size;
        //由于如果出现除不尽情况 最后的数据不会改变
        int count = (total % size == 0) ? num : num + 1;
        //定义方法进行分页提交操作
        CountDownLatch countDownLatch = new CountDownLatch((int) count);
        //定义方法进行分页提交操作
        for (int i = 0; i < count; i++) {
            //交给多线程管理
            ThreadPoolExecutor threadPoolExecutor = ProdThreadUtil.esImportThreadPool();
            threadPoolExecutor.execute(() -> {
                doImportPage(size, null, null);
                countDownLatch.countDown();
            });
        }
        try {
            //等待所有线程完成
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //设置定时任务开启时间
        start = new Date();
    }


    //
    private void doImportPage(Integer size, Date start, Date end) {
        //获得分页的数据
        Page<Prod> prodPage = new Page<>(1, size);
        //通过分页数据获得 prod数据
        List<Prod> prodList = prodService.selectDbData(prodPage, start, end);
        //将商品封装成 prodES数据
        //定义一个集合用于接收
        List<ProdEs> esLsit = new ArrayList<>(prodList.size());
        prodList.stream().forEach(prod -> {
            ProdEs prodEs = new ProdEs();
            //将prod数据拷贝到prodES中
            BeanUtil.copyProperties(prod, prodEs);
            //添加集合
            esLsit.add(prodEs);
            //将商品改为上架
            prod.setStatus(1);
        });
        //上传到es中
        propRepository.saveAll(esLsit);
        //更该数据库数据
        //商品变为上架
        prodService.updateBatchById(prodList);
    }

    /**
     * 增量导入： 定时任务处理
     * 每隔2个小时查询一次数据  查询是时间阶段的新增数据
     * 开始时间  结束时间
     */
    @Override
    //定时任务标签
    @Scheduled(initialDelay = 2 * 60 * 1000, fixedDelay = 2 * 60 * 1000)
    public void importUpdate() {
        //结束时间
        Date end = new Date();
        //获取商品总数
        int total = prodService.count(new LambdaQueryWrapper<Prod>()
                .eq(Prod::getStatus, 0)
                //查询这个时间范围类的
                .between(Prod::getCreateTime, start, end));
        //得到操作次数
        int num = total / size;
        //由于如果出现除不尽情况 最后的数据不会改变
        int count = (total % size == 0) ? num : num + 1;
        CountDownLatch countDownLatch = new CountDownLatch((int) count);
        //定义方法进行分页提交操作
        for (int i = 0; i < count; i++) {
            //交给多线程管理
            ThreadPoolExecutor threadPoolExecutor = ProdThreadUtil.esImportThreadPool();
            threadPoolExecutor.execute(() -> {
                doImportPage(size, start, end);
                countDownLatch.countDown();
            });
        }
        try {
            //等待所有线程完成
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //本次结束时间作为下一次的开始时间
        start = end;
    }


    /**
     * 快速导入  使用 消息队列mq操作
     * 下订单的时候： 库存 （mysql  es中）
     * 修改es中的库存  就使用快速导入
     * EsChange:        下单               退货
     * prodId   1   count -2            2
     * prodId   3   count -5            5
     * List<EsChange>  changeList
     */
    @Override
    @RabbitListener(queues = QueueConstant.ES_CHANGE_QUEUE)
    public void importQuick(Message message, Channel channel) {
        //获取消息
        String msg = new String(message.getBody());
        List<EsChange> esChangeList = JSON.parseArray(msg, EsChange.class);
        esChangeList.forEach(esChange -> {
            //首先获取原来的数量 然后修改
            ProdEs prodEs = propRepository.findById(esChange.getProdId()).get();
            Integer number = esChange.getNumber();
            //库存
            prodEs.setTotalStocks(prodEs.getTotalStocks() + number);
            //销量
            prodEs.setSoldNum(prodEs.getSoldNum() - number);
            propRepository.save(prodEs);
        });
        //签收消息
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 开启启动就执行一次
     *
     * @param args
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        importAll();
    }
}
